package org.yonggan.cmcc.report

import com.alibaba.fastjson.JSON
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, KafkaUtils}
import org.yonggan.cmcc.config.{ConfigConstant, KafkaParams}
import org.yonggan.cmcc.report.biz.RechargeTotalData
import org.yonggan.cmcc.util.Calculate


/**
  * 统计总量 数据
  *
  *     1. 交易金额总量
  *     2.充值成功总量
  *     3.充值失败总量
  *     4.充值总时长
  *
  */
object RechargeRptApplication {

  Logger.getLogger("org").setLevel(Level.WARN)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName("中国移动运营监控平台")
      .setMaster("local[*]")
      .set("spark.streaming.kafka.maxRatePerPartition", ConfigConstant.SPARK_MAXRATE) // 平滑限流策略
    .set("spark.streaming.stopGracefullyOnShutdown","true")
    val ssc = new StreamingContext(conf,ConfigConstant.SPARK_DURATION)

    val resouceDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](ConfigConstant.KAFKA_TOPIC, ConfigConstant.KAFKA_PARAMS, KafkaParams.kafkaOffset)
    )

    resouceDS.foreachRDD(rdd => {

      if (!rdd.isEmpty()) {
        //数据源头RDD
        val baseRDD: RDD[(List[Double],String,String,String,String)] = rdd.map(cr => { //ConsumerRecord[String, String]
          // 提取json 数组
          val json = cr.value()
          // 转换成为json对象
          val jsonObject = JSON.parseObject(json)

          val requestId = jsonObject.getString("requestId")
          val receiveNotifyTime = jsonObject.getString("receiveNotifyTime")

          //          (日期,小时，分钟， 订单总量 ， 成功订单量， 金额 ， 省份编码）
          val day = receiveNotifyTime.substring(0, 8)
          val hours = receiveNotifyTime.substring(8, 10)
          val min = receiveNotifyTime.substring(10, 12)

          // 是否是成功订单
          val bol = jsonObject.getString("bussinessRst").equalsIgnoreCase("0000")
          val fee = if (bol) jsonObject.getDouble("chargefee").toDouble else 0
          val succ = if (bol) 1 else 0
          // 通知支付结果
          val pcode = jsonObject.getString("provinceCode")

          // 支付时长
          val endTime = Calculate.dateStr2Long(receiveNotifyTime)
          val startTime = Calculate.dateStr2Long(requestId.substring(0,17))
          val castTime = Calculate.compare2Nums(endTime,startTime)((f1:Long,f2:Long)=>f1-f2)

          (List[Double](1, succ, fee, castTime),pcode,day,hours,min)
        })
        // 缓存
        baseRDD.cache()

        // 实时链式处理
        import org.yonggan.cmcc.report.RealTimeReportDateChain.function.HandlerFuntions._
        baseRDD.starHander

        RechargeTotalData.realTimeGrossData(baseRDD)

//        //存储偏移量
        val ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        KafkaParams.saveKafkaOffset(ranges)
      }
    })

    ssc.start()
    ssc.awaitTermination()
  }

}

